package com.jinghang.spark_base._010_RDD

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

object _040_map_mapPartition_foreach_foreachPartition {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("_050_mapPartition")
      .setMaster("local[1]")

    val sc = new SparkContext(conf)
    val arrayRDD = sc.parallelize(Array(1, 2, 3, 4, 5, 6, 7, 8, 9))

    //map函数每次处理一个/行数据
    arrayRDD.map(element => {
      element +1
    }).collect()
      .foreach(x => println(x))

    /*
        map() 和 foreach()的区别
          map()会生成新rdd
          foreach()返回值为unit

     */


    /*
        mapPartitions
          是transformation算子，
          每次处理一个分区中的数据，容易造成内存溢出问题，慎用
          返回新的rdd
     */
    arrayRDD.mapPartitions(elements => {
      //本地操作
      var result = new ArrayBuffer[Int]()
      elements.foreach(element => {
        result.+=(element)
      })
      result.iterator
    }).foreach(println)

    /*
        foreachPartition
           是action算子，
          每次处理一个分区中的数据，容易造成内存溢出问题，慎用
          一般在数据最终入库时候使用，批处理能提升性能
          返回unit
     */
    arrayRDD.foreachPartition(elements => {
      //本地操作一个分区中的所有元素
      var result = new ArrayBuffer[Int]()
      elements.foreach(element => {
        result.+=(element)
      })
      //result.iterator
      //保存数据的操作(mysql,hbase,redis)
      //把数组中的数组存入mysql中
    })
  }
}



